package cn.wangjie.spark.sources.file

import com.mysql.cj.protocol.a.NativeConstants.IntegerDataType
import org.apache.spark.sql.streaming.{OutputMode, StreamingQuery}
import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}
import org.apache.spark.sql.types.{IntegerType, StringType, StructType}

/**
 * 使用Structured Streaming从目录中读取文件数据：统计年龄小于25岁的人群的爱好排行榜
 */
object StructuredFileSource {
	
	def main(args: Array[String]): Unit = {
		
		// 1. 创建SparkSession实例对象
		val spark: SparkSession = SparkSession
			.builder()
			.appName(this.getClass.getSimpleName.stripSuffix("$"))
			.master("local[2]")
			// 设置SparkSQL Shuffle时分区数目
			.config("spark.sql.shuffle.partitions", "2")
			.getOrCreate()
		// 导入隐式转换和函数库
		import spark.implicits._
		import org.apache.spark.sql.functions._
		
		// TODO: 从文件数据源加载数据，本质就是监控目录中新文件，当做流式数据读取
		// 第一、定义Schema信息, 数据样本：honghong;28;running
		val schema: StructType = new StructType()
			.add("name", StringType, nullable = true)
			.add("age", IntegerType, nullable = true)
			.add("hobby", StringType, nullable = true)
		// 第二、加载文件
		val inputStreamDF: DataFrame = spark
			.readStream
			.schema(schema)      // Specify schema of the csv files
			.format("csv")
			.option("header", "false")
			.option("sep", ";")
			.load("file:///D:/datas/")
		
		// TODO: 按照具体业务分析数据：统计年龄小于25岁的人群的爱好排行榜
		val resultStreamDF: Dataset[Row] = inputStreamDF
    		// 过滤数据：年龄小于25
    		.filter($"age" < 25)
			// 按照爱好分组，统计count
    		.groupBy($"hobby").count()
			// 降序排序，按照count
    		.orderBy($"count".desc)
		
		// 4. 输出数据和启动启用
		val query: StreamingQuery = resultStreamDF
			.writeStream
			// 使用Complete完全模式输出，将ResultTable中所有数据进行输出
			.outputMode(OutputMode.Complete())
			.format("console")
			.option("numRows", "10")
			.option("truncate", "false")
			// 针对流式应用，需要启动执行
			.start()
		query.awaitTermination() // 当流式应用运行以后，正常情况一直运行，除非人为或程序异常终止，等待结束
		query.stop()
	}
	
}
